package org.example.mqstream.offsettracking;

import com.rabbitmq.stream.ByteCapacity;
import com.rabbitmq.stream.Environment;
import com.rabbitmq.stream.Producer;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

import static java.nio.charset.StandardCharsets.UTF_8;

public class OffsetTrackingSend {

    /**
     * 主程序入口
     * 本程序示例展示了如何使用MQ环境设置、创建流、发送和接收消息
     *
     * @param args 命令行参数
     * @throws InterruptedException 如果在等待消息确认时线程被中断
     */
    public static void main(String[] args) throws InterruptedException {
        // 使用try-with-resources确保环境资源在使用后能被正确关闭
        try (Environment environment = Environment.builder()
                .host("192.168.0.228")
                .port(5552)
                .username("mq")
                .password("123456")
                .build();) {
            // 定义流的名称
            String stream = "stream-offset-tracking-java";
            // 创建一个最大长度为1GB的流
            environment.streamCreator().stream(stream).maxLengthBytes(ByteCapacity.GB(1)).create();

            // 构建一个生产者，用于向指定流发送消息
            Producer producer = environment.producerBuilder().stream(stream).build();

            // 设置消息发送数量
            int messageCount = 100;
            // 创建一个计数器，用于等待所有消息的确认
            CountDownLatch confirmedLatch = new CountDownLatch(messageCount);
            System.out.printf("Publishing %d messages...%n", messageCount);
            // 使用IntStream发送指定数量的消息
            IntStream.range(0, messageCount).forEach(i -> {
                // 消息体，最后一消息使用"marker"作为标记，其他消息使用"hello"
                String body = i == messageCount - 1 ? "marker" : "hello";
                // 发送消息，并在消息确认时减少计数器
                producer.send(producer.messageBuilder().addData(body.getBytes(UTF_8)).build(), ctx -> {
                    if (ctx.isConfirmed()) {
                        confirmedLatch.countDown();
                    }
                });
            });

            // 等待所有消息被确认，最多等待60秒
            boolean completed = confirmedLatch.await(60, TimeUnit.SECONDS);
            // 输出消息确认结果
            System.out.printf("Messages confirmed: %b.%n", completed);
        }
    }

}
